package rmmk.algorithms.preprocessing;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.ArrayBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rmmk.algorithms.features.FeatureManager;
import rmmk.algorithms.knn.AbstractKnnClassifier;
import rmmk.algorithms.knn.KnnOpenCvImpl;
import rmmk.algorithms.knn.KnnOwnImpl;
import rmmk.algorithms.preprocessing.abstraction.IPreProcessor;
import rmmk.algorithms.preprocessing.abstraction.IWordFilter;
import rmmk.algorithms.preprocessing.config.DistanceMetric;
import rmmk.algorithms.preprocessing.config.KnnImpl;
import rmmk.algorithms.similarityMeasures.SimilarityMeasureManager;
import rmmk.framework.Utils;
import rmmk.framework.datasources.Document;

/**
 * Główna klasa frameworka. Posiada pola dla różnych ustawień takich jak: wersja
 * implementacji knn, wybrane filtry, algorytmy preprocessingu słów, rodzaj
 * metryki dystansu
 * 
 * @author marcin
 * 
 */
public class OperationManager {
	Logger logger = LoggerFactory.getLogger(OperationManager.class);

	KnnImpl knnImpl = KnnImpl.OwnImpl;
	DistanceMetric dm = DistanceMetric.Euclides;
	IWordFilter[] filters;
	IPreProcessor[] preProcessors;
	AbstractKnnClassifier akc = new KnnOwnImpl();
	FeatureManager featureManager = new FeatureManager();
	SimilarityMeasureManager similarityMeasureManager = new SimilarityMeasureManager();
	static public Integer nr = 0;

	public OperationManager() {

	}
	
	public void setK(int k){
		this.akc.setK(k);
	}

	public void preprocess(List<Document> documents) {
		for (IPreProcessor p : this.preProcessors) {
			logger.info("Preprocessing: " + p.getClass().getSimpleName());
			p.process(documents);
		}
	}

	public void filter(final List<Document> documents) {
		for (final IWordFilter f : this.filters) {
			logger.info("Preprocessing: " + f.getClass().getSimpleName());

			ArrayList<Thread> threads = new ArrayList<>();

			final int cores = Utils.getProcessorsCount();

			for (int i = 0; i < cores; ++i) {
				final int ii = i;
				threads.add(new Thread(new Runnable() {

					@Override
					public void run() {
						for (int j = ii; j < documents.size(); j += cores) {
							for (int w = 0; w < documents.get(j)
									.getExtractedWords().size(); ++w) {
								String wordOryg, word = documents.get(j)
										.getExtractedWords().get(w);
								wordOryg = word;

								word = f.processWord(word);

								if (!word.equals(wordOryg)) {
									documents.get(j).getExtractedWords()
											.remove(w);
									documents.get(j).getExtractedWords()
											.add(w, word);
								}
							}
						}
					}

				}));
			}

			for (Thread t : threads)
				try {
					t.join();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
	}

	public void teach(List<Document> documents) {
		//akc.setK(3);
		akc.teach(documents, featureManager, similarityMeasureManager);
	}

	public List<Document> predict(List<Document> documents) {
		logger.info("K: " +this.akc.getK());
		boolean singleThread = false;
		if(singleThread){
		
			for(Document d : documents)
				akc.predict(d, getDm());
			
			return documents;
		}
		else
		{
			int cores = Utils.getProcessorsCount();
			nr = 0;

			Vector<Thread> threads = new Vector<>();

			// List<Document> synchro = Collections.synchronizedList(documents);
			//ArrayBlockingQueue<Document> abq = new ArrayBlockingQueue<>(
			//		documents.size(), true, documents);
			
			ArrayList<ArrayList<Document>> documentLists = new ArrayList<>(cores);
			
			for(int i = 0; i < cores; ++i)
				documentLists.add(new ArrayList<Document>());
			//for(ArrayList<Document> al : documentLists)
			//	al = new ArrayList<>();
			
			for(int i = 0; i < documents.size(); ++i)
			{
				documentLists.get(i % cores).add(documents.get(i));
			}
			
			for (int i = 0; i < cores; ++i) {
				//Thread t = new Thread(new Processing2(i, cores, abq));
				Thread t = new Thread(new Processing3(i, cores, documentLists.get(i)));
				
				threads.add(t);

				t.start();
			}

			for (Thread t : threads)
				try {
					t.join();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			return documents;
		}
	}

	public KnnImpl getKnnImpl() {
		return knnImpl;
	}

	public void setKnnImpl(KnnImpl knnImpl) {
		if (this.knnImpl != knnImpl) {
			this.knnImpl = knnImpl;

			switch (knnImpl) {
			case OpenCV:
				this.akc = new KnnOpenCvImpl();
				break;
			case OwnImpl:
				this.akc = new KnnOwnImpl();
				break;
			}
		}
	}

	public DistanceMetric getDm() {
		return dm;
	}

	public void setDm(DistanceMetric dm) {
		this.dm = dm;
	}

	public IWordFilter[] getFilters() {
		return filters;
	}

	public void setFilters(IWordFilter[] filters) {
		this.filters = filters;
	}

	public IPreProcessor[] getPreProcessors() {
		return preProcessors;
	}

	public void setPreProcessors(IPreProcessor[] preProcessors) {
		this.preProcessors = preProcessors;
	}

	public FeatureManager getFeatureManager() {
		return featureManager;
	}

	public void setFeatureManager(FeatureManager featureManager) {
		this.featureManager = featureManager;
	}

	public SimilarityMeasureManager getSimilarityMeasureManager() {
		return similarityMeasureManager;
	}

	public void setSimilarityMeasureManager(
			SimilarityMeasureManager similarityMeasureManager) {
		this.similarityMeasureManager = similarityMeasureManager;
	}

	public AbstractKnnClassifier getAkc() {
		return akc;
	}

	class Processing implements Runnable {

		int shift, cores;
		private List<Document> documents;

		public Processing(int shift, int cores, List<Document> documents) {
			this.shift = shift;
			this.cores = cores;
			this.documents = documents;
		}

		@Override
		public void run() {
			/*
			 * for (int i = shift; i < documents.size(); i+=cores) {
			 * akc.predict(documents.get(i), dm); ++nr; if (nr % 100 == 0)
			 * logger.info("Document " + nr + " predicted"); }
			 */
			Iterator<Document> iter = null;
			while (true) {
				Document doc = null;
				synchronized (this.documents) {
					if (iter == null)
						iter = this.documents.iterator();
					if (iter.hasNext())
						doc = iter.next();
					else
						break;
				}
				akc.predict(doc, dm);
				synchronized (nr) {
					++nr;
					if (nr % 100 == 0)
						logger.info("Document " + nr + " predicted on thread: "
								+ this.shift);
				}
			}
		}

	}

	class Processing2 implements Runnable {

		int shift, cores;
		private ArrayBlockingQueue<Document> documents;

		public Processing2(int shift, int cores,
				ArrayBlockingQueue<Document> documents) {
			this.shift = shift;
			this.cores = cores;
			this.documents = documents;
		}

		@Override
		public void run() {
			logger.info("Jestem");
			while (!documents.isEmpty()) {
				try {
					Document doc = null;
					synchronized(documents)
					{
						doc = documents.take();
					}
					akc.predict(doc, dm);
					
					synchronized (nr) {
						++nr;
						if (nr % 100 == 0)
							logger.info("Document " + nr + " predicted on thread: "
									+ this.shift);
					}
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}
		}

	}
	
	private class Processing3 implements Runnable {

		int shift;//, cores;
		private List<Document> documents;

		public Processing3(int shift, int cores,
				List<Document> documents) {
			this.shift = shift;
			//this.cores = cores;
			this.documents = documents;
		}

		@Override
		public void run() {
			Iterator<Document> iter = documents.iterator();
			
			while(iter.hasNext())
			{
				akc.predict(iter.next(), dm);
				iter.remove();
				
				synchronized (nr) {
					++nr;
					if (nr % 100 == 0)
						logger.info("Document " + nr + " predicted on thread: "
								+ this.shift);
				}
			}
		}
		
	}

}
